package org.kafka.metrics;

import java.util.ArrayList;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import kafka.admin.ConsumerGroupCommand.ConsumerGroupCommandOptions;
import kafka.admin.ConsumerGroupCommand.KafkaConsumerGroupService;
//import kafka.admin.ConsumerGroupCommand.PartitionAssignmentState;
import scala.Option;
import scala.Tuple2;
import scala.collection.Seq;
import scala.collection.JavaConversions;

public class KafkaLagService_1002 {
    private static Logger log = LoggerFactory.getLogger(KafkaLagService_1002.class);
    public List<String> iniParams;
    private KafkaConsumerGroupService consumerGroupService;
    // val baseConfig: Array[String] = Array("--bootstrap-server",
    // kafkaSettings.address)

//    public List<String> listConsumergroup() {
//        List<String> options = new ArrayList<>(iniParams);
//        // options.add("--group");
//        ConsumerGroupCommandOptions cgOptions = new ConsumerGroupCommandOptions(options.toArray(new String[0]));
//        try {
//            consumerGroupService = new KafkaConsumerGroupService(cgOptions);
//            return JavaConversions.seqAsJavaList(consumerGroupService.listGroups());
//        } finally {
//            if (null != consumerGroupService) {
//                consumerGroupService.close();
//            }
//        }
//    }
//
//    public GroupInfo describe(String group) {
//        List<String> options = new ArrayList<>(iniParams);
//        options.add("--group");
//        options.add(group);
//        ConsumerGroupCommandOptions cgOptions = new ConsumerGroupCommandOptions(options.toArray(new String[0]));
//        System.out.println(cgOptions);
//        Tuple2<Option<String>, Option<Seq<PartitionAssignmentState>>> info;
//        try {
//            consumerGroupService = new KafkaConsumerGroupService(cgOptions);
//            info = consumerGroupService.describeGroup();
//            if (info._2().nonEmpty()) {
//                GroupInfo groupInfo = new GroupInfo().setState(info._1().get());
//                List<PartitionAssignmentStateBO> result = new ArrayList<>();
//                List<PartitionAssignmentState> list = JavaConversions.seqAsJavaList(info._2().get());
//                for (PartitionAssignmentState partitionAssignmentState : list) {
//                    PartitionAssignmentStateBO partInfo = new PartitionAssignmentStateBO();
//                    copy(partInfo, partitionAssignmentState);
//                    result.add(partInfo);
//                    log.info("pt{},lag{},offset{},logend{}", partInfo.partition, partInfo.lag, partInfo.offset,
//                            partInfo.logEndOffset);
//                }
//                groupInfo.setPatitions(result);
//                return groupInfo;
//            } else {
//                return new GroupInfo().setState(info._1().get());
//            }
//        } finally {
//            if (null != consumerGroupService) {
//                consumerGroupService.close();
//            }
//        }
//    }
//
//    public static void copy(PartitionAssignmentStateBO to, PartitionAssignmentState from) {
//        if (null == to || from == null) {
//            throw new IllegalStateException("from or to is null");
//        }
//        to.setClientID(from.clientId().get());
//        to.setGroup(from.group());
//        to.setTopic(from.topic().get());
//        to.setConsumerID(from.consumerId().get());
//        to.setLag((Long) from.lag().get());
//        to.setLogEndOffset((Long) from.logEndOffset().get());
//        to.setOffset((Long) from.offset().get());
//        to.setPartition((Integer) from.partition().get());
//        if (from.coordinator().nonEmpty()) {
//            to.setCoordinator("" + from.coordinator().get().id());
//        }
//    }
}

class PartitionAssignmentStateBO {
    public String group;
    public String coordinator;
    public String topic;
    public int partition;
    public long offset;
    public long lag;
    public long logEndOffset;
    public String consumerID;
    public String clientID;

    public String getGroup() {
        return group;
    }

    public PartitionAssignmentStateBO setGroup(String group) {
        this.group = group;
        return this;
    }

    public String getCoordinator() {
        return coordinator;
    }

    public PartitionAssignmentStateBO setCoordinator(String coordinator) {
        this.coordinator = coordinator;
        return this;
    }

    public String getTopic() {
        return topic;
    }

    public PartitionAssignmentStateBO setTopic(String topic) {
        this.topic = topic;
        return this;
    }

    public int getPartition() {
        return partition;
    }

    public PartitionAssignmentStateBO setPartition(int partition) {
        this.partition = partition;
        return this;
    }

    public long getOffset() {
        return offset;
    }

    public PartitionAssignmentStateBO setOffset(long offset) {
        this.offset = offset;
        return this;
    }

    public long getLag() {
        return lag;
    }

    public PartitionAssignmentStateBO setLag(long lag) {
        this.lag = lag;
        return this;
    }

    public long getLogEndOffset() {
        return logEndOffset;
    }

    public PartitionAssignmentStateBO setLogEndOffset(long logEndOffset) {
        this.logEndOffset = logEndOffset;
        return this;
    }

    public String getConsumerID() {
        return consumerID;
    }

    public PartitionAssignmentStateBO setConsumerID(String consumerID) {
        this.consumerID = consumerID;
        return this;
    }

    public String getClientID() {
        return clientID;
    }

    public PartitionAssignmentStateBO setClientID(String clientID) {
        this.clientID = clientID;
        return this;
    }

}

class GroupInfo {
    public String state;
    public List<PartitionAssignmentStateBO> patitions;

    public String getState() {
        return state;
    }

    public GroupInfo setState(String state) {
        this.state = state;
        return this;
    }

    public List<PartitionAssignmentStateBO> getPatitions() {
        return patitions;
    }

    public GroupInfo setPatitions(List<PartitionAssignmentStateBO> patitions) {
        this.patitions = patitions;
        return this;
    }

}